# -*- coding: utf-8 -*-
import json
import os
import signal
import time
from concurrent import futures

import grpc
from kafka import KafkaProducer

from common.conf import KAFKA_BOOTSTRAP_SERVERS, SERVICE_TOPIC_REGISTER, \
    SERVICE_TOPIC_UNREGISTER
from common.conf import SERVICE_TIME_INTERVAL
from common.utils.id_util import uuid
# auto import with build_proto.py
from user_module.handler.menu import MenuHandler
from user_module.handler.role import RoleHandler
from user_module.handler.user import UserHandler
from user_module.rpc import user_pb2_grpc
# end auto

_SERVICE_ID = uuid()
_MODULE = 'user'
_HOST = 'localhost'
_PORT = '8080'

content = {
    'service_id': _SERVICE_ID,
    'service_name': _MODULE,
    'host': _HOST,
    'port': _PORT
}
content = json.dumps(content).encode()
user_producer = KafkaProducer(bootstrap_servers=KAFKA_BOOTSTRAP_SERVERS)


def serve():
    grpc_server = grpc.server(futures.ThreadPoolExecutor(max_workers=4))

    # 从这里开始是build_proto.py 自动添加
    user_pb2_grpc.add_MenuServicer_to_server(MenuHandler(), grpc_server)
    user_pb2_grpc.add_RoleServicer_to_server(RoleHandler(), grpc_server)
    user_pb2_grpc.add_UserServicer_to_server(UserHandler(), grpc_server)
    # 自动添加结束

    grpc_server.add_insecure_port(_HOST + ':' + _PORT)
    grpc_server.start()

    try:
        while True:
            future = user_producer.send(SERVICE_TOPIC_REGISTER, content)
            future.get(timeout=120)
            time.sleep(SERVICE_TIME_INTERVAL)

    except KeyboardInterrupt:
        unregister_content = {
            'service_id': _SERVICE_ID,
            'service_name': _MODULE,
        }
        unregister_content = json.dumps(unregister_content).encode()
        future = user_producer.send(SERVICE_TOPIC_UNREGISTER, unregister_content)
        future.get(timeout=120)
        grpc_server.stop(0)
        os.kill(os.getpid(), signal.SIGKILL)


if __name__ == '__main__':
    serve()
